package james.gan.remoting.netty.pool;

import java.net.SocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import james.gan.common.Request;
import james.gan.common.Response;
import james.gan.remoting.api.RemotingException;
import james.gan.remoting.api.RemotingFutureDispatchMonitor;
import james.gan.remoting.api.RemotingFutureDispatchMonitor.RemotingFuture;
import james.gan.serialization.api.Serialization;

public class NettyClientProxy {

	private static final Logger logger = LoggerFactory.getLogger(NettyClientProxy.class);

	public final SocketAddress socketAddress;

	private final Serialization serialization;

	private static final RemotingFutureDispatchMonitor reMonitor = new RemotingFutureDispatchMonitor();

	final EventLoopGroup group = new NioEventLoopGroup();

	private Channel channel;

	public NettyClientProxy(SocketAddress socketAddress, Serialization serialization) throws InterruptedException {
		super();
		this.socketAddress = socketAddress;
		this.serialization = serialization;
		initChannel();
	}

	private void initChannel() throws InterruptedException {
		// 客户端
		Bootstrap bootstrap = new Bootstrap();

		bootstrap.group(group).channel(NioSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO))
				.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true);
		ChannelHandler[] handlers = new ChannelHandler[] { new IdleStateHandler(0, 1, 0, TimeUnit.MINUTES),
				new NettyDecoder(Response.class), new NettyEncoder(), new NettyClientHandler() };

		bootstrap.handler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(handlers);
			}
		});
		channel = bootstrap.connect(socketAddress).sync().addListener(f -> {
			if (f.isSuccess()) {
				logger.debug("Connect {} success", socketAddress);
			} else {
				logger.debug("Connect {} failed", socketAddress);
			}
		}).channel();
	}

	public void close() {
		group.shutdownGracefully(0, 5000, TimeUnit.SECONDS);
		if (channel != null && channel.isOpen()) {
			channel.close();
		}
	}

	public RemotingFuture send(Request request) throws RemotingException {
		this.channel.writeAndFlush(request);
		return reMonitor.getResult(request.getId());
	}
	
	class NettyDecoder extends ByteToMessageDecoder {

		private Class<?> clzz;

		public NettyDecoder(Class<?> clzz) {
			this.clzz = clzz;
		}

		@Override
		protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
			if (in.readableBytes() < 4) {
	            return;
	        }
	        in.markReaderIndex();
	        int dataLength = in.readInt();
	        if (dataLength < 0) {
	            ctx.close();
	        }
	        if (in.readableBytes() < dataLength) {
	            in.resetReaderIndex();
	            return;	// fix 1024k buffer splice limix
	        }
	        byte[] data = new byte[dataLength];
	        in.readBytes(data);

	        Object obj = serialization.deserialize(data, clzz);
	        out.add(obj);
		}
	}

	class NettyEncoder extends MessageToByteEncoder<Request> {

		@Override
		protected void encode(ChannelHandlerContext ctx, Request msg, ByteBuf out) throws Exception {
			byte[] data = serialization.serialize(msg);
			out.writeInt(data.length);
			out.writeBytes(data);
		}
	}

	@ChannelHandler.Sharable
	class NettyClientHandler extends ChannelInboundHandlerAdapter {

		private final ByteBuf HEARTBEAT_SEQUENCE = Unpooled
				.unreleasableBuffer(Unpooled.copiedBuffer("Heartbeat", CharsetUtil.UTF_8));

		@Override
		public void channelActive(ChannelHandlerContext ctx) throws Exception {
			super.channelActive(ctx);
		}

		@Override
		public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
			if (msg instanceof Response) {
				reMonitor.setResult((Response) msg);
			} else {
				logger.error("Server exception with error responseType :{}", msg.getClass());
			}
		}

		@Override
		public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
			if (evt instanceof IdleStateEvent) {
				IdleState state = ((IdleStateEvent) evt).state();
				if (state == IdleState.WRITER_IDLE) {
					System.out.println("发送心跳");
					ctx.writeAndFlush(HEARTBEAT_SEQUENCE.duplicate());
				}
			} else {
				super.userEventTriggered(ctx, evt);
			}
		}
	}
}
